package com.bianmaba.jobs.collection.processor;

import com.alibaba.fastjson.JSON;
import com.bianmaba.beans.datacollection.ExecuteLog;
import com.bianmaba.beans.datamanagement.DataTable;
import com.bianmaba.dc.bean.CollectionData;
import com.bianmaba.dc.bean.DataType;
import com.bianmaba.dc.bean.RedisKeys;
import com.bianmaba.dc.bean.TaskStatus;
import com.bianmaba.services.datacollection.ExecuteLogRecorder;
import com.bianmaba.services.datacollection.ITaskInstanceService;
import com.bianmaba.services.datamanagement.DataTableService;
import lombok.extern.log4j.Log4j2;
import org.quartz.Job;
import org.quartz.JobDataMap;
import org.quartz.JobExecutionContext;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.ListOperations;
import org.springframework.transaction.annotation.Transactional;

import java.util.List;

@Transactional
@Log4j2
public abstract class AbstractProcessor implements Job {
    @Autowired
    protected ExecuteLogRecorder logWriter;
    @Autowired
    protected DataTableService dataTableService;
    @Autowired
    protected ITaskInstanceService taskInstanceService;
    @Autowired
    protected ListOperations<String, Object> redisListOperations;

    @Override
    public void execute(JobExecutionContext context) {
        JobDataMap jobDatas = context.getMergedJobDataMap();
        CollectionData datas = (CollectionData) jobDatas.get("datas");
        String taskInstanceId = datas.getTaskInstanceId();

        logWriter.put(ExecuteLog.of(taskInstanceId, "处理器：开始处理采集数据，taskInstanceId=" + taskInstanceId));
        List<CollectionData.Table> tables = datas.getTables();
        for (CollectionData.Table table : tables) {
            List<CollectionData.Row> rows = table.getRows();
            DataTable dataTable = DataTable.of(datas.getProjectId(), datas.getInterfaceType(), datas.getTaskId(), table.getDataType(), rows);
            try {
                if (table.getDataType().equalsIgnoreCase(DataType.SOURCE_DATA)) {
                    mergeSourceDatas(taskInstanceId, dataTable);
                } else if (table.getDataType().equalsIgnoreCase(DataType.PROJECT_STATISTICS)) {
                    mergeStatistics(taskInstanceId, dataTable);
                } else if (table.getDataType().equalsIgnoreCase(DataType.WORKFORCE_STATISTICS)) {
                    mergeStatistics(taskInstanceId, dataTable);
                } else if (table.getDataType().equalsIgnoreCase(DataType.DEMAND_STATISTICS)) {
                    mergeStatistics(taskInstanceId, dataTable);
                }
                redisListOperations.leftPush(RedisKeys.COLLECTION_DATA_PARSED, JSON.toJSONString(dataTable));
                logWriter.put(ExecuteLog.of(taskInstanceId, "处理器：采集数据处理完成，taskInstanceId=" + taskInstanceId));
                taskInstanceService.updateStatus(taskInstanceId, TaskStatus.SUCCESSFUL.getIndex());
            } catch (Exception e) {
                String err = "处理器：保存数据出现异常，taskInstanceId=" + taskInstanceId;
                log.error(err, e);
                logWriter.put(ExecuteLog.of(taskInstanceId, err));
                taskInstanceService.updateStatus(taskInstanceId, TaskStatus.EXCEPTIONAL.getIndex());
                return;
            }
        }
    }

    public abstract void mergeStatistics(String taskInstanceId, DataTable table);

    public abstract void mergeSourceDatas(String taskInstanceId, DataTable table);
}
